import getopt
import sys
import traceback
from datetime import date, datetime, timedelta

import pandas as pd
from es_pandas import es_pandas

from air_web.config.config import config
from air_web.data_platform import sql_engine
from air_web.web_flask.dal.base_dal import EsBaseDal


def check_predict_data(es_dal, org_no, start_date, end_date):
    sql = f"""select on5, data_time, p_total
              from {config['POWERLOAD_PREDICT_VIRTUAL_TABLE']} 
              where on5={org_no} and on7=9999999 and type_id=99999
                and data_time >= '{start_date}' and data_time < '{end_date}'
              order by on5, data_time
           """
    mysql_df = sql_engine.query(sql)

    rules = [
        ("on5", "query", "=", org_no),
        ("type_id", "query", "=", "99999"),
        ("on7", "query", "=", "9999999"),
        ("data_time", "query", ">=", start_date),
        ("data_time", "query", "<", end_date),
    ]
    index_name = config["POWERLOAD_PREDICT_VIRTUAL_IDX"]
    es_df = es_dal.query_dataframe(
        rules,
        index_name,
        doc_time_field="data_time",
        source=["on5", "data_time", "p_total"],
    )

    merge_df = mysql_df.merge(
        es_df, on=["on5", "data_time"], how="left", suffixes=("_sql", "_es")
    )
    merge_df["diff"] = merge_df["p_total_es"] - merge_df["p_total_sql"]

    # todo 7位数存入mysql，最后一位会四舍五入，所以取10判断
    not_less_than_1 = merge_df[
        (merge_df["diff"] > 10) | (merge_df["diff"] < -10)
    ]

    if not_less_than_1.empty:
        print(f"mysql和es数据完全相同:{org_no}")
    else:
        print(f"mysql和es数据不完全相同:{org_no}")
        print(not_less_than_1)


def get_es_virtual_date(es_pd, c_id, cons_no, es_time_list):
    """
    查询es中后处理的聚合索引结果和虚拟用户计算结果
    Args:
        c_id: 虚拟用户c_id
        cons_no: 虚拟用户cons_no （c_cons表）
        es_time_list: 结果索引中要查询的时间点

    Returns:

    """
    virtual_output_df = pd.DataFrame()
    virtual_input_df = pd.DataFrame()
    date_list = set()
    time_list = list()
    for dt in es_time_list:
        # 使用 datetime.strftime 函数将 datetime 对象转换为日期字符串
        date_str = datetime.strftime(dt, "%Y-%m-%dT00:00:00")
        time_str = datetime.strftime(dt, "%Y-%m-%d %H:%M:%S")
        # 将日期字符串添加到日期列表中
        date_list.add(date_str)
        time_list.append(time_str)
    date_list = list(date_list)
    query_rule = {
        "query": {
            "bool": {
                "must": [
                    {"term": {"c_id": str(c_id)}},
                    {"terms": {"data_time": time_list}},
                ]
            }
        },
        # "_source": ["data_time", "p_total", "p_kt"]
    }
    try:
        virtual_output_df = es_pd.to_pandas(
            config["POWERLOAD_VIRTUAL_IDX"], query_rule=query_rule
        )
    except:
        print(f"es查询{config['POWERLOAD_VIRTUAL_IDX']}失败")
        print(traceback.format_exc())
    if not virtual_output_df.empty:
        virtual_output_df = virtual_output_df[["data_time", "p_total", "p_kt"]]
        virtual_output_df["data_time"] = pd.to_datetime(
            virtual_output_df["data_time"]
        )
    input_query_rule = {
        "query": {
            "bool": {
                "must": [
                    {"term": {"id": str(cons_no)}},
                    {"terms": {"data_date": date_list}},
                ]
            }
        },
        # "_source": ["data_date", "value"]
    }
    # print(input_query_rule)
    try:
        virtual_input_df = es_pd.to_pandas(
            config["POWER96BYDAY_VIRTUAL"], query_rule=input_query_rule
        )
    except:
        print(f"es查询{config['POWER96BYDAY_VIRTUAL']}失败")
        print(traceback.format_exc())
    # 转变格式
    if not virtual_input_df.empty:
        virtual_input_df = virtual_input_df[["data_date", "value"]]
        # virtual_input_df['data_time'] = virtual_input_df.apply(
        #     lambda row: [row['date_date'] + pd.Timedelta(minutes=15 * i) for i in range(len(row['value']))], axis=1)
        # 将 value 列拆分为 96 个值
        virtual_input_df[
            ["value{}".format(i) for i in range(96)]
        ] = virtual_input_df["value"].str.split(",", expand=True)
        # 删除原始的 value 列
        df = virtual_input_df.drop("value", axis=1)
        # 创建一个新的 date_time 列
        df["data_time"] = pd.to_datetime(df["data_date"])
        # 删除原始的 date_date 列
        df = df.drop("data_date", axis=1)
        # 对于每个值，生成一个新行，其中 date_time 增加 15 分钟
        new_df = pd.DataFrame()
        for i in range(96):
            temp_df = df.copy()
            temp_df["data_time"] = temp_df["data_time"] + timedelta(
                minutes=15 * i
            )
            temp_df["p_total"] = temp_df["value{}".format(i)].astype(float)
            new_df = new_df.append(temp_df)
        # 删除临时列
        new_df = new_df.drop(["value{}".format(i) for i in range(96)], axis=1)
        virtual_input_df = new_df.reset_index(drop=True)

        # 使用 explode 函数将 value 和 data_time 列拆分为多行
        virtual_input_df = virtual_input_df[
            virtual_input_df["data_time"].isin(es_time_list)
        ]

    return virtual_input_df, virtual_output_df


def check_virtual_data(es_pd, org_no, start_str, end_str):
    """
    检查同步结果正确性
    Args:
        org_no: 同步的org编号
        start_str: 开始时间
        end_str: 结束时间

    Returns: True or False, 同时打印报告

    """
    print_str = f"on5级别同步核查 org_no:{org_no} start_date:{start_str} end_data {end_str}\n"
    cons_no = f"99999_{org_no}"
    c_id_query = f"select DISTINCT c_id,cons_name from c_cons where cons_no = '{cons_no}'"
    c_id_df = sql_engine.query(c_id_query)
    if c_id_df.empty:
        print_str += f"c_cons表中查不到{org_no} 对应的全行业 c_id"
        print(print_str)
        return False
    c_id = c_id_df.iloc[0]["c_id"]
    cons_name = c_id_df.iloc[0]["cons_name"]
    print_str += f"cons_name: {cons_name}\n"
    time_list = []
    s_date = datetime.strptime(start_str, "%Y-%m-%d")
    e_date = datetime.strptime(end_str, "%Y-%m-%d")
    days = (e_date - s_date).days
    if days > 10 or days < 0:
        print("请检查时间区间，时间间隔仅限10天内")
        return False

    # 遍历每一天
    current_date = s_date
    while current_date <= e_date:
        # 添加 6 点、12 点和 18 点时刻
        for hour in [6, 12, 18]:
            dt = current_date.replace(hour=hour)
            time_list.append(dt.strftime("%Y-%m-%d %H:%M:%S"))
        # 移动到下一天
        current_date += timedelta(days=1)
    time_filter_str = ",".join([f"'{i}'" for i in time_list])
    virtual_table = config["POWERLOAD_VIRTUAL_TABLE"]  # 虚拟用户计算结果表
    virtual_query = (
        f"select data_time, p_total, p_kt from {virtual_table} where c_id = {c_id} "
        f"and data_time in ({time_filter_str})"
    )
    virtual_df = sql_engine.query(virtual_query)

    org_15min_table = config["ORGNO_TYPEID_15MIN"]  # 15min聚合结果表
    org_15min_query = (
        f"select data_time, p_total_sum as p_total, p_kt_sum as p_kt from {org_15min_table} "
        f"where org_no = {org_no} and type_id = {0} and data_time in ({time_filter_str})"
    )
    org_15min_df = sql_engine.query(org_15min_query)

    if org_15min_df.empty:
        print_str += f"15min 表查询为空，所以请检查时间和org_no"
        print(print_str)
        return False
    merged_df = pd.merge(
        org_15min_df,
        virtual_df,
        on="data_time",
        how="outer",
        suffixes=("_15min", "_virtual"),
    )
    not_equal = merged_df[
        (merged_df["p_total_15min"] - merged_df["p_total_virtual"]).abs() > 10
    ]  # 不匹配的列(插值大于10)

    query_time_list = (
        merged_df["data_time"].dt.strftime("%Y-%m-%d %H:%M:%S").to_list()
    )
    lost_time_list = list(
        set(time_list) - set(query_time_list)
    )  # 在15min表和虚拟用户结果都没有查到的时间点

    if lost_time_list:
        lost_time_list.sort()
        print_str += (
            f"以下时间点结果在15min表和虚拟结果表中都没查到,疑似是没有计算到可忽略 {lost_time_list}\n"
        )
    if not_equal.empty:  # 查到的时间点内 p_total都是匹配的
        print_str += f"全行业单用户聚合总负荷与虚拟用户计算总负荷均相同(差值10以内)\n"

        merged_df["data_time"] = pd.to_datetime(merged_df["data_time"])
        # 使用布尔索引筛选出 data_time 列值为 12:00 的行
        result = merged_df[merged_df["data_time"].dt.hour == 12]
        print_str += f"结果抽样：{result.to_dict('record')}\n"
        print(print_str)
        return True
    # 查到的时间点内 p_total不匹配
    print_str += f"全行业单用户聚合总负荷与虚拟用户计算总负荷6:00,12:00,18:00抽样点存在差异,做进一步查询\n"
    # es_time_list = lost_time_list + not_equal["data_time"].to_list()   # 查询虚拟用户结果缺失的时间点和不一致的时间点
    es_time_list = not_equal[
        "data_time"
    ].to_list()  # 查询不一致的时间点, 默认15min表是完整的，lost_time_list es中肯定也查不到
    virtual_input_df, virtual_output_df = get_es_virtual_date(
        es_pd, c_id, cons_no, es_time_list
    )

    if not virtual_input_df.empty:
        # 比较后聚合save_es索引正确性
        input_15min_merge = pd.merge(
            org_15min_df,
            virtual_input_df,
            on="data_time",
            how="outer",
            suffixes=("_15min", "_input"),
        )
        input_15min_not_equal = input_15min_merge[
            (
                input_15min_merge["p_total_15min"]
                - input_15min_merge["p_total_input"]
            ).abs()
            > 10
        ]  # 后处理聚合索引不匹配的行
        # 情况一： 后处理聚合结果不一致（包含es中不存在）
        if input_15min_not_equal.empty:
            print_str += (
                f"索引{config['POWER96BYDAY_VIRTUAL']}核验数据p_total与15min表匹配\n"
            )
        else:
            print_str += (
                f"索引{config['POWER96BYDAY_VIRTUAL']}核验数据p_total与15min表存在不匹配，抽样异常点为"
                f"{input_15min_not_equal['data_time'].dt.strftime('%Y-%m-%d %H:%M:%S').to_list()}\n"
            )
            result = input_15min_not_equal[
                input_15min_not_equal["data_time"].dt.hour == 12
            ]
            print_str += f"结果抽样：{result.to_dict('record')}\n"
    else:
        print_str += "后处理聚合结果索引查询为空或查询结果报错，请检查\n"

    if not virtual_output_df.empty:
        output_15min_merge = pd.merge(
            org_15min_df,
            virtual_output_df,
            on="data_time",
            how="outer",
            suffixes=("_15min", "_output"),
        )
        output_15min_not_equal = output_15min_merge[
            (
                output_15min_merge["p_total_15min"]
                - output_15min_merge["p_total_output"]
            ).abs()
            > 10
        ]  # 虚拟用户计算结果不匹配的行

        miss_virtual = not_equal[not_equal["p_total_virtual"].isnull()]
        miss_output = output_15min_not_equal[
            output_15min_not_equal["p_total_output"].isnull()
        ]

        # 情况二： 虚拟用户计算没完成或没同步完
        if not miss_output.empty:
            print_str += (
                f"预估虚拟用户没有计算完成，15min表有但虚拟用户计算索引没有查询到的抽样时间点 "
                f"{miss_output['data_time'].dt.strftime('%Y-%m-%d %H:%M:%S').to_list()}\n"
            )
        if not miss_virtual.empty:
            sub_list = list(
                set(miss_virtual["data_time"].to_list())
                - set(miss_output["data_time"].to_list())
            )
            if sub_list:
                print_str += f"计算完成但是没有同步至mysql，抽样时间点 {sub_list}\n"
            else:
                print_str += (
                    f"预估虚拟用户没有计算完成, 15min表有但虚拟用户mysql表中没有查询到的抽样时间点"
                    f" {miss_virtual['data_time'].dt.strftime('%Y-%m-%d %H:%M:%S').to_list()}\n"
                )

        # 情况三： 虚拟用户计算不正确
        not_equal = not_equal[~not_equal["p_total_virtual"].isnull()]
        output_15min_not_equal = output_15min_not_equal[
            ~output_15min_not_equal["p_total_output"].isnull()
        ]
        if not output_15min_not_equal.empty:
            print_str += (
                f"es虚拟用户计算结果错误。其中15min表与虚拟用户计算索引不匹配的抽样时间点 "
                f"{output_15min_not_equal['data_time'].dt.strftime('%Y-%m-%d %H:%M:%S').to_list()}\n"
            )
            result = output_15min_not_equal[
                output_15min_not_equal["data_time"].dt.hour == 12
            ]
            print_str += f"结果抽样：{result.to_dict('record')}\n"
        if not not_equal.empty:
            print_str += (
                f"同步的mysql表计算结果错误。15min表有但虚拟用户mysql表中不匹配的抽样时间点 "
                f"{not_equal['data_time'].dt.strftime('%Y-%m-%d %H:%M:%S').to_list()}\n"
            )
            result = not_equal[not_equal["data_time"].dt.hour == 12]
            print_str += f"结果抽样：{result.to_dict('record')}\n"
    else:
        print_str += "虚拟用户计算结果索引查询为空或查询结果报错，请检查\n"
    print(print_str)
    return False


if __name__ == "__main__":
    ep = es_pandas(config["ES_HOST"])
    es_dal = EsBaseDal(config["ES_HOST"])
    opts, args = getopt.getopt(sys.argv[1:], "s:e:i:p:")
    start_date = (date.today() - timedelta(days=5)).strftime("%Y-%m-%d")
    end_date = date.today().strftime("%Y-%m-%d")
    pre_end_date = (date.today() + timedelta(days=3)).strftime("%Y-%m-%d")
    input_orgno = None  # on5 或者 on7
    for opt, val in opts:
        if opt == "-s":
            start_date = val
        elif opt == "-e":
            end_date = val
        elif opt == "-p":
            pre_end_date = val
        elif opt == "-i":
            input_orgno = int(val)
    if not input_orgno:
        print("必须使用-i参数指定on5或on7编号，-s 开始时间 -e 结束时间 选填")
    else:
        check_virtual_data(ep, input_orgno, start_date, end_date)
        check_predict_data(es_dal, input_orgno, start_date, pre_end_date)
